//
// Created by jiao on 2021/4/15.
//

#include "YHThreadHandler.h"

#include <cstdint>
#include <cstring>
#include <cstdlib>
#include <unistd.h>
#include <pthread.h>
#include <sys/eventfd.h>
#include "jni.h"
#include <cerrno>

static const int EPOLL_MAX_EVENTS = 16;
//退出循环的标志
static const int EXIT_IDENT = -2;

void *YHThreadHandler::threadFunc(void *p) {
    ((YHThreadHandler *) p)->backgroundRun();
    return NULL;
}

YHThreadHandler::~YHThreadHandler() {
    YQ_HANDLER_LOGD("YHThreadHandler  折构方法");
}


YHThreadHandler::YHThreadHandler(YqHandlerCallBackFunc *callBack) : mCallBack(callBack) {
    mEventFd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (mEventFd < 0) {
        YH_HANDLER_LOGE("YHThreadHandler  Could not create epoll instance: %s", strerror(errno));
    }
    pthread_mutex_init(&mLooper_mutex_, NULL);
    mEpollFd = epoll_create(8);
    struct epoll_event eventItem{};
    memset(&eventItem, 0, sizeof(epoll_event)); // zero out unused members of data field union
    eventItem.events = EPOLLIN;
    eventItem.data.fd = mEventFd;
    int result = epoll_ctl(mEpollFd, EPOLL_CTL_ADD, mEventFd, &eventItem);
    if (result != 0) {
        YH_HANDLER_LOGE("YHThreadHandler  Could not add wake event fd to epoll instance: %s",
                        strerror(errno));
    }

    pthread_t tidp_;
    pthread_create(&tidp_, NULL, threadFunc, this);
}

void YHThreadHandler::backgroundRun() {
    //当前需要处理的任务
    std::vector<std::pair<int, void *>> curQueue;
    //是否退出循环
    bool isExit = false;

    while (true) {
        while (mQueue.empty()) {
            mPolling = true;
            struct epoll_event eventItems[EPOLL_MAX_EVENTS];
//            YQ_HANDLER_LOGD("epoll_wait-->start");

            epoll_wait(mEpollFd, eventItems, EPOLL_MAX_EVENTS, -1);
            // No longer idling.
            mPolling = false;
//            YQ_HANDLER_LOGD("epoll_wait-->stop  eventCount:%d", eventCount);
        }

        pthread_mutex_lock(&mLooper_mutex_);
        awoken();
        if (!mQueue.empty()) {
            for (auto data:mQueue) {
                if (data.first == EXIT_IDENT) {
                    isExit = true;
                    break;
                }
                curQueue.push_back(data);
            }
        }
        mQueue.clear();
        pthread_mutex_unlock(&mLooper_mutex_);
        for (auto data:curQueue) {
            mCallBack(data.first, data.second);
        }
        curQueue.clear();
        if (isExit) {
            destroy();
            break;
        }
    }

}

void YHThreadHandler::sendMessage(int ident, void *data) {
    if (isQuit)return;
    pthread_mutex_lock(&mLooper_mutex_);
    mQueue.emplace_back(ident, data);
    pthread_mutex_unlock(&mLooper_mutex_);
    if (mPolling) {
        wake();
    }
}

void YHThreadHandler::removeMessage(int ident) {
    pthread_mutex_lock(&mLooper_mutex_);
    std::vector<std::pair<int, void *>>::iterator iter;
    for (iter = mQueue.begin(); iter != mQueue.end();) {
        auto tmp = *iter;
        if (tmp.first == ident) {
            iter = mQueue.erase(iter);
        } else {
            ++iter;
        }
    }
    pthread_mutex_unlock(&mLooper_mutex_);
}

void YHThreadHandler::removeAll() {
    if (isQuit)return;
    pthread_mutex_lock(&mLooper_mutex_);
    mQueue.clear();
    pthread_mutex_unlock(&mLooper_mutex_);

}

void YHThreadHandler::destroy() {
    YQ_HANDLER_LOGD("YHThreadHandler  destroy");

    isQuit = true;
    pthread_mutex_destroy(&mLooper_mutex_);
    mQueue.clear();
    close(mEventFd);
    mEventFd = -1;
    if (mEpollFd >= 0) {
        close(mEpollFd);
    }
    mCallBack = nullptr;
    delete this;
}

void YHThreadHandler::wake() {
    YQ_HANDLER_LOGD("YHThreadHandler  %p ~ wake", this);
    uint64_t inc = 1;
    ssize_t nWrite = TEMP_FAILURE_RETRY(write(mEventFd, &inc, sizeof(uint64_t)));
    if (nWrite != sizeof(uint64_t)) {
        if (errno != EAGAIN) {
            YH_HANDLER_LOGE("YHThreadHandler  Could not write wake signal to fd %d: %s",
                            mEventFd, strerror(errno));
        }
    }
}

void YHThreadHandler::awoken() {
    YQ_HANDLER_LOGD("YHThreadHandler  %p ~ awoken", this);
    uint64_t counter;
    TEMP_FAILURE_RETRY(read(mEventFd, &counter, sizeof(uint64_t)));
}

void YHThreadHandler::quit() {
    sendMessage(EXIT_IDENT, NULL);
}

